package org.tao.lightningmq.broker.event.spi.listenr;

import com.alibaba.fastjson2.JSON;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.tao.lightningmq.broker.cache.CommonCache;
import org.tao.lightningmq.broker.event.model.ConsumeMsgEvent;
import org.tao.lightningmq.broker.model.ConsumeQueueConsumeReqModel;
import org.tao.lightningmq.broker.rebalance.ConsumerInstance;
import org.tao.lightningmq.common.dto.*;
import org.tao.lightningmq.common.enums.BrokerResponseCode;
import org.tao.lightningmq.common.event.Listener;
import org.tao.lightningmq.common.tcp.TcpMsg;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;

/**
 * @Author lht
 * @date 2025/1/20 - 14:35
 * @description: 消费者拉取消息事件监听器
 */
public class ConsumeMsgListener implements Listener<ConsumeMsgEvent> {

    private static final Logger LOGGER = LoggerFactory.getLogger(ConsumeMsgListener.class);

    @Override
    public void onReceive(ConsumeMsgEvent event) throws Exception {
        ConsumeMsgReqDTO consumeMsgReqDTO = event.getConsumeMsgReqDTO();
        String currentReqId = consumeMsgReqDTO.getReqId();
        ConsumeMsgBaseRespDTO consumeMsgBaseRespDTO = new ConsumeMsgBaseRespDTO();
        List<ConsumeMsgRespDTO> consumeMsgRespDTOS = new ArrayList<>();
        consumeMsgBaseRespDTO.setConsumeMsgRespDTOList(consumeMsgRespDTOS);
        consumeMsgBaseRespDTO.setMsgId(event.getMsgId());
        Map<String, List<ConsumerInstance>> consumeGroupMap =
                CommonCache.getConsumeHoldMap().get(consumeMsgReqDTO.getTopic());

        //有可能当前消费组还没经过第一轮重平衡，因此不会那么快消费到数据,所以要通知客户端，目前服务端还没将队列分配好
        if (consumeGroupMap == null) {
            LOGGER.debug("consumeGroupMap为空，没有消费到数据, topic 为: {}", consumeMsgReqDTO.getTopic());
            //直接返回空数据
            event.getContext().writeAndFlush(new TcpMsg(BrokerResponseCode.CONSUME_MSG_RESP.getCode(),
                    JSON.toJSONBytes(consumeMsgBaseRespDTO)));
            return;
        }
        List<ConsumerInstance> consumerInstances =
                consumeGroupMap.get(consumeMsgReqDTO.getConsumeGroup());
        if (consumerInstances == null || consumerInstances.isEmpty()) {
            LOGGER.warn("consumerInstances为空，没有消费到数据");
            //直接返回空数据
            event.getContext().writeAndFlush(new TcpMsg(BrokerResponseCode.CONSUME_MSG_RESP.getCode(),
                    JSON.toJSONBytes(consumeMsgBaseRespDTO)));
            return;
        }

        // 可以获取到数据
        consumerInstances.stream()
                .filter(instance -> instance.getConsumerReqId().equals(currentReqId))
                .forEach(instance -> {
                    int batchSize = instance.getBatchSize();
                    ConsumeQueueConsumeReqModel consumeQueueConsumeReqModel = new ConsumeQueueConsumeReqModel();
                    consumeQueueConsumeReqModel.setTopic(consumeMsgReqDTO.getTopic());
                    consumeQueueConsumeReqModel.setConsumeGroup(instance.getConsumeGroup());
                    for (Integer queueId : instance.getQueueIdSet()) {
                        if (batchSize == 0) break;
                        consumeQueueConsumeReqModel.setBatchSize(batchSize);
                        consumeQueueConsumeReqModel.setQueueId(queueId);
                        List<ConsumeMsgCommitLogDTO> commitLogContentList =
                                CommonCache.getConsumeQueueConsumeHandler().consume(consumeQueueConsumeReqModel);
                        if (commitLogContentList == null || commitLogContentList.isEmpty()) continue;
                        ConsumeMsgRespDTO consumeMsgRespDTO = new ConsumeMsgRespDTO();
                        consumeMsgRespDTO.setQueueId(queueId);
                        consumeMsgRespDTO.setCommitLogContentList(commitLogContentList);
                        consumeMsgRespDTOS.add(consumeMsgRespDTO);
                        batchSize -= commitLogContentList.size();
                    }
        });
        LOGGER.info("读取到的数据为: {}", JSON.toJSON(consumeMsgBaseRespDTO));
        byte[] body = JSON.toJSONBytes(consumeMsgBaseRespDTO);
        TcpMsg respMsg = new TcpMsg(BrokerResponseCode.CONSUME_MSG_RESP.getCode(),
                body);
        event.getContext().writeAndFlush(respMsg);

    }

}
